from mrjob.job import MRJob, MRStep
from mrjob.protocol import TextProtocol
import heapq


class UserValue(MRJob):

    # OUTPUT_PROTOCOL = TextProtocol

    def mapper_step1(self, _, line):
        lineinfo = line.strip().split(',')
        if len(lineinfo) == 6:
            if lineinfo[3] == 'buy':
                yield lineinfo[0], 1
            else:
                yield lineinfo[0], 0

    def combiner_step1(self, key, values):
        yield key, sum(values)

    def reducer_step1(self, key, values):
        yield None, (sum(values), key)

    def reducer_step2(self, _, values):

        for count, key in sorted(values, reverse=True):
            yield key, count



    def steps(self):
        return [
            MRStep(mapper=self.mapper_step1,
                   combiner=self.combiner_step1,
                   reducer=self.reducer_step1),

            MRStep(reducer=self.reducer_step2)
        ]


if __name__ == '__main__':
    UserValue.run()

